﻿using System;
using System.Collections.Generic;
using System.IO.Pipes;
using System.Linq;
using System.Threading;
using SimplePipeHelper.Common;

namespace SimplePipeHelper.Client
{
    internal class PipeClientWriteHelper : IDisposable
    {
        private readonly Action _onPipeClosed; //管道服务端关闭
        private readonly Action<string> _onErrorMessage; //错误信息

        private readonly Queue<byte[]> _msgQueue = new Queue<byte[]>();
        private readonly Semaphore _msgSemaphore = new Semaphore(0, int.MaxValue);

        private NamedPipeClientStream _pipeClient;
        private PipeAsyncState _asyncState;

        private Thread _threadWorker;
        private bool _threadWorking;
        private readonly object _threadWorkingLockHelper = new object();

        private bool IsThreadWorking
        {
            get
            {
                bool ret;
                lock (_threadWorkingLockHelper)
                {
                    ret = _threadWorking;
                }

                return ret;
            }
            set
            {
                lock (_threadWorkingLockHelper)
                {
                    _threadWorking = value;
                }
            }
        }

        public PipeClientWriteHelper(
            Action onPipeClosed,
            Action<string> onErrorMessage)
        {
            _onPipeClosed = onPipeClosed;
            _onErrorMessage = onErrorMessage;
        }

        public void Start(NamedPipeClientStream pipeClient)
        {
            _pipeClient = pipeClient;
            if (pipeClient == null)
            {
                return;
            }

            _asyncState = new PipeAsyncState()
            {
                PipeStream = _pipeClient,
                EventHandle = new ManualResetEvent(false)
            };

            IsThreadWorking = true;
            _threadWorker = new Thread(WriteThreadFunc) {IsBackground = true};
            _threadWorker.Start();
        }

        public void Stop()
        {
            //发送终止信号
            IsThreadWorking = false;
            _asyncState?.SetEvent();

            //终止线程
            _threadWorker?.Join();
            _threadWorker = null;

            //释放同步资源
            _asyncState?.Dispose();
            _asyncState = null;

            if (HasMessage())
            {
                ClearMessage();
            }
        }

        #region 消息队列操作

        public void AddMessage(byte[] msg)
        {
            lock (_msgQueue)
            {
                _msgQueue.Enqueue(msg);
                _msgSemaphore.Release();
            }
        }

        private bool HasMessage()
        {
            bool ret;
            lock (_msgQueue)
            {
                ret = _msgQueue.Any();
            }

            return ret;
        }

        private byte[] PickMessage()
        {
            byte[] msg = null;
            lock (_msgQueue)
            {
                if (_msgQueue.Count > 0)
                {
                    msg = _msgQueue.Peek();
                    _msgQueue.Dequeue();
                }
            }

            return msg;
        }

        private void ClearMessage()
        {
            lock (_msgQueue)
            {
                _msgQueue.Clear();
            }
        }

        #endregion

        private void WriteThreadFunc()
        {
            while (IsThreadWorking)
            {
                if (_pipeClient == null || !_pipeClient.IsConnected || _asyncState?.EventHandle == null)
                {
                    break;
                }

                if (!_msgSemaphore.WaitOne(1))
                {
                    continue;
                }

                Thread.Sleep(1);

                var msg = PickMessage();

                try
                {
                    _asyncState.Buffer = msg;
                    _asyncState.ResetEvent();

                    _pipeClient.BeginWrite(msg, 0,
                        msg.Length, OnPipeWriteCallback, _asyncState);
                    _asyncState.EventHandle.WaitOne();
                }
                catch (ObjectDisposedException e)
                {
                    _onErrorMessage?.Invoke(e.Message);

                    //管道已关闭
                    _onPipeClosed?.Invoke();
                    break;
                }
                catch (InvalidOperationException e)
                {
                    _onErrorMessage?.Invoke(e.Message);

                    //管道已断开，正在等待连接
                    _onPipeClosed?.Invoke();
                    break;
                }
                catch (Exception e)
                {
                    _onErrorMessage?.Invoke(e.Message);
                    _onPipeClosed?.Invoke();
                    break;
                }
            }
        }

        private void OnPipeWriteCallback(IAsyncResult ar)
        {
            var state = (PipeAsyncState) ar.AsyncState;
            if (state == null)
            {
                throw new NullReferenceException();
            }

            state.PipeStream.EndWrite(ar);
            state.PipeStream.Flush();
            state.SetEvent();
        }

        #region IDisposable Support

        //检测冗余调用
        private bool _disposedValue;

        private void Dispose(bool disposing)
        {
            if (!_disposedValue)
            {
                if (disposing)
                {
                    //释放托管状态(托管对象)。
                    Stop();
                }

                //释放未托管的资源(未托管的对象)并在以下内容中替代终结器。
                //将大型字段设置为 null。

                _disposedValue = true;
            }
        }

        ~PipeClientWriteHelper()
        {
            Dispose(false);
        }

        // 添加此代码以正确实现可处置模式。
        void IDisposable.Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        #endregion
    }
}
